package net.bwie.realtime.douyin.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.douyin.bean.PayViewBean;
import net.bwie.realtime.douyin.function.PayViewBeanMapFunction;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/26 22:37:19
 * @version: 1.0
 */
public class Pay_flow_into_clickhouse {
    public static void main(String[] args) throws Exception {
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //3接受 支付流
        DataStream<String> pay_Stream = KafkaUtil.consumerKafka(env, "pay-log");


//        pay_Stream.print();

        DataStream<String> handle = handle(pay_Stream);



        env.execute("Dwd_Split_flow");

    }

    private static DataStream<String> handle(DataStream<String> payStream) {

        KeyedStream<String, String> keyedStream = payStream.keyBy(json -> JSON.parseObject(json).getString("order_id"));

        SingleOutputStreamOperator<PayViewBean> beanStream = keyedStream.map(new PayViewBeanMapFunction());
//
        beanStream.print();

        return null;
    }
}
